//package com.cnebula.dataprocess.sqoop.job.comsumer;
//
//import java.util.HashMap;
//import java.util.HashSet;
//import java.util.Map;
//import java.util.Set;
//
//import javax.annotation.Resource;
//
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener;
//import org.apache.rocketmq.spring.starter.core.RocketMQListener;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//import org.springframework.stereotype.Service;
//
//import com.aqlu.rocketmq.demo.consumer.OrderPaidEventConsumer;
//import com.aqlu.rocketmq.demo.domain.OrderPaidEvent;
//import com.cnebula.dataprocess.common.sqoop.DataSourceNode;
//import com.cnebula.dataprocess.rocketmq.starter.AbstractRocketMqConsumer;
//import com.cnebula.dataprocess.rocketmq.starter.constants.RocketMqTopic;
//import com.cnebula.dataprocess.sqoop.job.config.RocketConsumerConfig;
//import com.cnebula.dataprocess.sqoop.job.config.SqoopConfig;
//import com.cnebula.dataprocess.sqoop.job.model.FromSqoopJobConstants;
//import com.cnebula.dataprocess.sqoop.job.model.LinkConfig;
//import com.cnebula.dataprocess.sqoop.job.model.LspMLink;
//import com.cnebula.dataprocess.sqoop.job.model.ToSqoopJobConstants;
//import com.cnebula.dataprocess.sqoop.job.service.SqoopService;
//
//
//@Component
//public class SqoopJobConsumer extends AbstractRocketMqConsumer<RocketMqTopic, DataSourceNode> {
//
//	@Resource
//	RocketConsumerConfig consumerConf;
//
//	@Resource
//	SqoopConfig sqoopConf;
//
//	@Autowired
//	SqoopService sqoopService;
//
//	@Override
//	public Map<String, Set<String>> subscribeTopicTags() {
//		Map<String, Set<String>> map = new HashMap<>();
//		Set<String> tags = new HashSet<>();
//		tags.add(consumerConf.getTag());
//		map.put(consumerConf.getTopic(), tags);
//		return map;
//	}
//
//	@Override
//	public String getConsumerGroup() {
//		return consumerConf.getGroup();
//	}
//
//	@Override
//	public boolean consumeMsg(DataSourceNode node, MessageExt msg) {
//		
//		logger.info("receive sqoop job");
//
//		String dsType = node.getDbType();
//
//		LinkConfig fromLinkConfig = null;
//
//		if (dsType.equalsIgnoreCase(LinkConstants.TYPE_MYSQL)) {
//			fromLinkConfig = getMysqlLinkConfig(node);
//		}
//
//		if (fromLinkConfig == null) {
//			// TODO send error message
//		}
//
//		LinkConfig toLinkConfig = new LinkConfig();
//		toLinkConfig.setUrl(sqoopConf.getHadoopNamenode());
//		toLinkConfig.setType(LinkConstants.TYPE_HDFS);
//		toLinkConfig.setHadoopConfDir(sqoopConf.getHadoopConfDir());
//		
//		Map<String, String> fromJobConfig = new HashMap<String, String>();
//		fromJobConfig.put(FromSqoopJobConstants.JDBC_SQL, node.getSql());
//		fromJobConfig.put(FromSqoopJobConstants.JDBC_PARTITIONCOLUMN, node.getPartition());
//
//		Map<String, String> toJobConfig = new HashMap<String, String>();
//		toJobConfig.put(ToSqoopJobConstants.HDFS_OUTPUTDIRECTORY, sqoopConf.getJobDir());
//
//		try {
//			LspMLink fromLink = sqoopService.createMLink(fromLinkConfig);
//			LspMLink toLink = sqoopService.createMLink(toLinkConfig);
//			sqoopService.submitJob(fromLink, toLink, fromJobConfig, toJobConfig);
//		} catch (Exception e) {
//		}
//
//		return true;
//	}
//
//	private LinkConfig getMysqlLinkConfig(DataSourceNode node) {
//
//		String host = node.getHost();
//		int port = node.getPort();
//		String encoding = node.getCharset();
//		String username = node.getUsername();
//		String password = node.getPassword();
//		String databaseName = node.getDatabase();
//
//		StringBuilder url = new StringBuilder("jdbc:mysql://");
//		url.append(host).append(":").append(port);
//		url.append("/");
//		url.append(databaseName);
//
//		if (encoding != null && !encoding.isEmpty()) {
//			url.append("?");
//			url.append("useUnicode=true");
//			url.append("&");
//			url.append("characterEncoding=" + encoding);
//		}
//
//		LinkConfig linkConfig = new LinkConfig();
//		linkConfig.setUrl(url.toString());
//		linkConfig.setDriver("com.mysql.jdbc.Driver");
//		linkConfig.setType(LinkConstants.TYPE_MYSQL);
//		linkConfig.setUserName(username);
//		linkConfig.setPassword(password);
//
//		return linkConfig;
//	}
//
//}
